/**
 * Copyright 2016 vip.com.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
 * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations under the License.
 * </p>
 **/

package spring.cloud.tasks.api.service.impl.statistics.analyzer;

import com.google.common.base.Strings;
import org.apache.commons.lang3.StringUtils;
import spring.cloud.tasks.api.domain.JobStatistics;
import spring.cloud.tasks.api.domain.NameSpaceConfiguration;
import spring.cloud.tasks.api.utils.TaskNodePath;
import spring.cloud.tasks.api.zookeeper.ZooKeeper;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author timmy.hu
 */
public class JobStatisticsAnalyzer {

    private final Map<String/** {jobname}-{domain} */, JobStatistics> jobMap = new ConcurrentHashMap<>();

    public static int getProcessCountAllTime(ZooKeeper zooKeeper, String jobName) {
        String processCountOfThisJobAllTimeStr = zooKeeper.getData(TaskNodePath.getProcessCountPath(jobName));
        return StringUtils.isBlank(processCountOfThisJobAllTimeStr) ? 0 : Integer.parseInt(processCountOfThisJobAllTimeStr);
    }

    public static int getErrorCountAllTime(ZooKeeper zooKeeper, String jobName) {
        String errorCountOfThisJobAllTimeStr = zooKeeper.getData(TaskNodePath.getErrorCountPath(jobName));
        return StringUtils.isBlank(errorCountOfThisJobAllTimeStr) ? 0 : Integer.parseInt(errorCountOfThisJobAllTimeStr);
    }

    public JobStatistics analyze(ZooKeeper zooKeeper, String job,
                                 NameSpaceConfiguration config, ExecutorInfoAnalyzer executorInfoAnalyzer) throws Exception {
        String jobDomainKey = job + "-" + config.getNameSpace();
        JobStatistics jobStatistics = jobMap.get(jobDomainKey);
        if (jobStatistics == null) {
            jobStatistics = new JobStatistics(job, config.getNameSpace(), config.getNameSpace());
            jobMap.put(jobDomainKey, jobStatistics);
        }

        String jobDegree = zooKeeper.getData(TaskNodePath.getConfigNodePath(job, "jobDegree"));
        if (Strings.isNullOrEmpty(jobDegree)) {
            jobDegree = "0";
        }
        jobStatistics.setJobDegree(Integer.parseInt(jobDegree));

        int processCountOfThisJobAllTime = getProcessCountAllTime(zooKeeper, job);
        int errorCountOfThisJobAllTime = getErrorCountAllTime(zooKeeper, job);

        // loadLevel of this job
        int loadLevel = Integer.parseInt(zooKeeper.getData(TaskNodePath.getConfigNodePath(job, "loadLevel")));
        int shardingTotalCount = Integer.parseInt(zooKeeper.getData(TaskNodePath.getConfigNodePath(job, "shardingTotalCount")));
        List<String> servers = null;
        if (zooKeeper.checkExists(TaskNodePath.getExecutorsNodePath(job))) {
            servers = zooKeeper.getChildren(TaskNodePath.getExecutorsNodePath(job));
            executorInfoAnalyzer.analyzeServer(zooKeeper, servers, job, config.getNameSpace(), config, loadLevel, jobStatistics);
        }
        // local-mode job = server count(regardless server status)
        {
            jobStatistics.setTotalLoadLevel(loadLevel * shardingTotalCount);
        }
        jobStatistics.setErrorCountOfAllTime(errorCountOfThisJobAllTime);
        jobStatistics.setProcessCountOfAllTime(processCountOfThisJobAllTime);
        jobMap.put(jobDomainKey, jobStatistics);
        return jobStatistics;
    }

    public Map<String, JobStatistics> getJobMap() {
        return jobMap;
    }

    public List<JobStatistics> getJobList() {
        return new ArrayList<JobStatistics>(jobMap.values());
    }
}
